package cn.doitedu.day05

import cn.doitedu.day01.utils.SparkUtil

/**
 * @Date 22.4.3
 * @Created by HANGGE
 * @Description
 */
object C18_转换算子_Aggregate {
  def main(args: Array[String]): Unit = {
    val sc = SparkUtil.getSc
    //  [1,2,3]  [4,5,6]
    val rdd1 = sc.parallelize(List(1, 2, 3, 4, 5,6),2)

    /**
     * 所有数据的均值
     *   每个分区的总和  和  总个数
     *   (x1,x2)=>x1+x2  x1+x2的结果返回给x1
     *   (0,0) [1]  => (1,1)  (tp,v)=>(tp._1+v , tp._2+1)
     *   将每个分区的总和  和 总个数  汇总
     *   (6,  3)
     *   (15 ,3)    (tp1,tp2)=>(tp1._1+tp2._1 , tp1._2+tp2._2)
     */

   /*   160
    10 + +21*/
    /**
     *     rdd1 的并行度问题
     *     master ==>  local       1
     *     master ==>  local[N]    n
     *     master ==>  local[*]    所有可用核数
     *     指定了分区数据    指定的
      */

    // 柯里化的方法
    //  (zeroValue: U)参数列表一   默认值  初始值
    //    分区内的聚合逻辑        各个分区结果的处理逻辑
    //  (seqOp: (U, T) => U, combOp: (U, U) => U)参数列表二
    //  : U  返回值
    /**
     *  + 1) +2 )+3 =  (6,3)
     *  (0,0) => (1,1)
     *  (1,1) =>(3,2)
     *  (3,2) = >(6,3)
     *  + 4) +5)+6 =   (15,3)
     *   (21 ,6)  (Int ,Int)
     *    21/6
     */
      // 行动算子  返回值 [总和 , 个数]
    val res: Int = rdd1.aggregate(10)(_ + _, _ + _)
    // 求所有数据的平均值
    val res2 = rdd1.aggregate((0, 0))(
      (tp, v) => (tp._1 + v, tp._2 + 1)
      , (tp1, tp2) => (tp1._1 + tp2._1, tp1._2 + tp2._2)
    )
    println(res2._1.toDouble/res2._2.toDouble)
    println(res)
  }

}
